package com.hbase_mapreduce.read;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.QualifierFilter;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.util.Arrays;

public class MyClazzCount {


    public static void main(String[] args) throws Exception {

        // 解析项目中的配置文件
        Configuration conf = new Configuration(true);

        //工具类帮我们把-D 等等的属性直接set到conf，会留下commandOptions
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] othargs = parser.getRemainingArgs();

        System.out.println(Arrays.toString(othargs));

        //让框架知道是windows异构平台运行
        conf.set("hbase.zookeeper.quorum", "node02,node03,node04");
        conf.set("mapreduce.app-submission.cross-platform", "true");
        // 配置文件中的 mapreduce.framework.name 属性默认是 yarn，也就是集群模式
        conf.set("mapreduce.framework.name", "local");

        Job job = Job.getInstance(conf);

        // 本地模式提交到yarn执行，需要setJar
        // job.setJar("D:\\bigdata_project\\bigdata_test\\target\\bigdata_test-1.0-SNAPSHOT.jar");

        //必须必须写的,反推这个类属于哪个jar包
        job.setJarByClass(MyClazzCount.class);

        job.setJobName("hbase_wc");

        // 输出路径，命令行传入
        Path outPath = new Path(othargs[0]);

        // 如果路径存在，先删除path file
        if (outPath.getFileSystem(conf).exists(outPath)) {
            outPath.getFileSystem(conf).delete(outPath, true);
        }

        // 从Hbase的哪个表读取哪些数据
        Scan scan = new Scan();

        // 只查询班级列
        QualifierFilter qualifierFilter = new QualifierFilter(CompareOperator.EQUAL,
                new BinaryComparator(Bytes.toBytes("clazz")));

        // 获取student表的前100条数据，按照clazz分组，计算每个班级的学生人数
        PageFilter pageFilter = new PageFilter(100);

        FilterList filterList = new FilterList();
        filterList.addFilter(qualifierFilter);
        filterList.addFilter(pageFilter);

        scan.setFilter(filterList);

        TableName tableName = TableName.valueOf("api_test", "student");

        // 设置读取的表，和scan的规则
        TableMapReduceUtil.initTableMapperJob(tableName,
                scan,
                MyMapper.class,
                Text.class,
                IntWritable.class,
                job);

        // 只能设置一个输出路径
        TextOutputFormat.setOutputPath(job, outPath);
        job.setReducerClass(MyReducer.class);
        // reduce输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);


        // Submit the job, then poll for progress until the job is complete
        job.waitForCompletion(true);

    }

}
